package com.atguigu.flink.chapter07;

import com.atguigu.flink.chapter05.Source.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/20 14:06
 */
public class Flink21_Process_Timer {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 8888)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {   // 指定如何从数据中提取事件时间
                                    @Override
                                    public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                        return element.getTs() * 1000L; // 转换成 毫秒
                                    }
                                })
                );

        OutputTag<WaterSensor> outputTag = new OutputTag<WaterSensor>("late") {
        };

        SingleOutputStreamOperator<String> resultDS = sensorDS
                .keyBy(sensor -> sensor.getId())
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                        // 获取的是数据的事件时间
//                        Long timestamp = ctx.timestamp();
//                        String currentKey = ctx.getCurrentKey();
                        // TODO 定时器 API
                        // 1.注册定时器
//                        ctx.timerService().registerEventTimeTimer();
//                        ctx.timerService().registerProcessingTimeTimer();
                        // 2.删除定时器
//                        ctx.timerService().deleteEventTimeTimer();
//                        ctx.timerService().deleteProcessingTimeTimer();
                        // 3.获取当前时间的进展
//                        ctx.timerService().currentWatermark();
//                        ctx.timerService().currentProcessingTime();

                        // TODO 定时器使用
//                        long timerTs = ctx.timerService().currentProcessingTime() + 5000L;
                        long timerTs = 5000L;
                        System.out.println("注册定时器，ts=" + timerTs);
//                        ctx.timerService().registerProcessingTimeTimer(timerTs);
                        ctx.timerService().registerEventTimeTimer(timerTs);
                    }

                    /**
                     * 定时器触发：表示 时间进展到了 注册的那个时间
                     * @param timestamp 触发的时间，也就是 注册的那个时间
                     * @param ctx   上下文
                     * @param out   采集器
                     * @throws Exception
                     */
                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        System.out.println("定时器触发了，timestamp=" + timestamp);
                    }
                });


        resultDS.print("result");


        env.execute();
    }

}
/*
    1、定时器注册？
        => 每调用一次 register方法，都会 new一个定时器对象，并把定时器对象，放入一个 队列 里
        => 这个 队列 add 对象，会去重 => 如果重复注册同一个时间的定时器，只会放入一个对象
        => 去重，范围是同一分组，组内去重

    2、定时器触发？
        watermark >= 注册的时间

    思考：为什么定时器延迟了 1s触发？
        因为 watermark = 最大事件时间 - 乱序程度 - 1ms， 减去了 1ms
        例如，乱序程度设置3s，注册了一个 10s的定时器，
        当 eventtime = 13s时， watermark = 13s - 3s -1ms = 9999ms，小于 注册的时间 10s，无法触发。
        只有当 eventtime = 14s， watermark = 14s -3 - 1ms = 10999ms， 大于 注册的时间 10s，可以触发。



    eventTimeTimersQueue.add(new TimerHeapInternalTimer<>(time, (K) keyContext.getCurrentKey(), namespace));
        => 每调用一次 register方法，都会 new一个定时器对象
        => 把定时器对象，放入一个 队列 里
        => 这个 队列 add对象，会去重 => 如果重复注册同一个时间的定时器，只会放入一个对象

    timer.getTimestamp() <= time
        => timer.getTimestamp() 就是定时器注册的时间
        => time = watermark.getTimestamp()，就是 watermark
        => 注册的时间 <= watermark
 */